﻿#region COPYRIGHT© 2009-2012 Phillip Clark. All rights reserved.

// For licensing information see License.txt (MIT style licensing).

#endregion

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Diagnostics.Contracts;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using FlitBit.Cache.Config;
using FlitBit.Core;
using FlitBit.Core.Parallel;

namespace FlitBit.Cache
{
	/// <summary>
	///   Indicator flags used to return a status for an EvictionCandidate
	/// </summary>
	[Flags]
	internal enum ShouldEvict
	{
		None = 0,
		Yes = 1,
		No = 1 << 1,
		Expired = 1 << 2 | Yes,
		Revised = 1 << 3 | Yes,
		Cleared = 1 << 4 | No
	}

	/// <summary>
	///   Local memory cache implementation that keeps only the most recently used
	///   items in memory.
	/// </summary>
	/// <typeparam name="TKey">key type K</typeparam>
	/// <typeparam name="TValue">item type V</typeparam>
	public class MruLocalMemoryCache<TKey, TValue> : Disposable, ICache<TKey, TValue>
		where TValue : class
	{
		readonly ConcurrentDictionary<TKey, Item> _cache = new ConcurrentDictionary<TKey, Item>();

		readonly int _cacheDepth;
		readonly ICacheEvictionManager _evictionManager;
		readonly ConcurrentQueue<EvictionCandidate> _evictionQueue = new ConcurrentQueue<EvictionCandidate>();
		readonly int _evictionSynchronizedThreshold;
		readonly int _evictionThreshold;
		readonly int _evictionsPerSynchronizedWorker;
		readonly ConcurrentQueue<EvictionCandidate> _lruQueue = new ConcurrentQueue<EvictionCandidate>();
		int _activeEvictorCount;
		readonly TimeSpan _cacheItemLifespan;
		ICacheEvictionMonitor _evictionMonitor;
		int _evictorCount;
		int _revision;
		ICacheStatsCollector _stats;
		string _what;

		/// <summary>
		///   Creates a new instance using the default cache options.
		/// </summary>
		/// <param name="name">the cache's name</param>
		public MruLocalMemoryCache(string name)
			: this(name, null, null, null)
		{
			Contract.Requires<ArgumentNullException>(name != null);
			Contract.Requires<ArgumentException>(name.Length > 0);
		}

		/// <summary>
		///   Creates a new instance using the default cache options.
		/// </summary>
		/// <param name="name">the cache's name</param>
		/// <param name="evictionManager">eviction manager</param>
		public MruLocalMemoryCache(string name, ICacheEvictionManager evictionManager)
			: this(name, null, null, evictionManager)
		{
			Contract.Requires<ArgumentNullException>(name != null);
			Contract.Requires<ArgumentException>(name.Length > 0);
		}

		/// <summary>
		///   Creates a new instance.
		/// </summary>
		/// <param name="name">the cache's name</param>
		/// <param name="options">options</param>
		public MruLocalMemoryCache(string name, MruCacheOptions options)
			: this(name, options, null, null)
		{
			Contract.Requires<ArgumentNullException>(name != null);
			Contract.Requires<ArgumentException>(name.Length > 0);
		}

		/// <summary>
		///   Creates a new instance.
		/// </summary>
		/// <param name="name">the cache's name</param>
		/// <param name="options">options</param>
		/// <param name="evictionManager">eviction manager</param>
		public MruLocalMemoryCache(string name,
			MruCacheOptions options, ICacheEvictionManager evictionManager)
			: this(name, options, null, evictionManager)
		{
			Contract.Requires<ArgumentNullException>(name != null);
			Contract.Requires<ArgumentException>(name.Length > 0);
		}

		/// <summary>
		///   Creates a new instance.
		/// </summary>
		/// <param name="name">the cache's name</param>
		/// <param name="stats">statistics collector</param>
		/// <param name="options">options</param>
		public MruLocalMemoryCache(string name,
			MruCacheOptions options, ICacheStatsCollector stats)
			: this(name, options, stats, null)
		{
			Contract.Requires<ArgumentNullException>(name != null);
			Contract.Requires<ArgumentException>(name.Length > 0);
		}

		/// <summary>
		///   Creates a new instance.
		/// </summary>
		/// <param name="name">the cache's name</param>
		/// <param name="stats">statistics collector</param>
		/// <param name="options">options</param>
		/// <param name="evictionManager">eviction manager</param>
		public MruLocalMemoryCache(string name,
			MruCacheOptions options, ICacheStatsCollector stats, ICacheEvictionManager evictionManager)
		{
			Contract.Requires<ArgumentNullException>(name != null);
			Contract.Requires<ArgumentException>(name.Length > 0);

			this.Name = name;

			var resolvedOptions = options ?? CacheConfigSection.Current.NamedOrDefaultOptions<TValue>(name);
			_stats = stats ?? CreateDefaultStatsCollector();
			_cacheDepth = resolvedOptions.CacheDepth;
			_evictionThreshold = Convert.ToInt32(Math.Round(_cacheDepth * resolvedOptions.ConcurrentEvictionFactor));
			_evictionSynchronizedThreshold =
				Convert.ToInt32(Math.Round(_evictionThreshold * resolvedOptions.SynchronizedEvictionFactor));
			_evictionsPerSynchronizedWorker = resolvedOptions.EvictionsPerSynchronizedWorker;
			_cacheItemLifespan = resolvedOptions.CacheItemLifespan;
			_evictionManager = evictionManager ?? FactoryProvider.Factory.CreateInstance<ICacheEvictionManager>();
			var contextKeys = resolvedOptions.ContextKeys;
			var enumerable = contextKeys as string[] ?? contextKeys.ToArray();
			if (enumerable.Any())
			{
				_evictionMonitor = new ReactiveCacheEvictionMonitor<TKey>(enumerable, EvictionNotificationCallback);
				_evictionManager.AddEvictionMonitor(_evictionMonitor);
			}
		}

		/// <summary>
		///   Gets the cache's depth.
		/// </summary>
		public int CacheDepth { get { return _cacheDepth; } }

		/// <summary>
		///   Gets the number of items in the cache.
		/// </summary>
		public int Count { get { return _cache.Count; } }

		/// <summary>
		///   Gets the threshold at which concurrent eviction threads are
		///   launched.
		/// </summary>
		/// <remarks>
		///   When the threshold is exceeded (cache count)
		///   concurrent eviction workers will be started to remove the
		///   least-recently-used items from the cache.
		/// </remarks>
		public decimal EvictionThreshold { get { return _evictionThreshold; } }

		/// <summary>
		///   Number of evictions processed per synchronized eviction event.
		/// </summary>
		/// <remarks>
		///   If the synchronization threshold is exceeded then a calling
		///   thread is borrowed to process evictions. This setting
		///   indicates the max number of evictions the borrowed thread
		///   should process before returning the thread to the caller.
		/// </remarks>
		public int EvictionsPerSynchronizedWorker { get { return _evictionsPerSynchronizedWorker; } }

		/// <summary>
		///   Gets the threshold at which the caller's thread will be borrowed
		///   to process evictions.
		/// </summary>
		/// <remarks>
		///   When the threshold is exceeded the caller's thread will be borrowed
		///   to process evictions. This threshold prevents the cache from growing
		///   when it encounters heavy writes of unique items greatly exceeding
		///   the eviction threshold. It limits the cache's growth when there are
		///   more writer threads than eviction workers.
		/// </remarks>
		public decimal SynchronizedEvictionThreshold { get { return _evictionSynchronizedThreshold; } }

		internal ICacheStatsCollector Stats { get { return _stats; } }

		string What
		{
			get
			{
				return Util.NonBlockingLazyInitializeVolatile(ref _what, () => this.GetType()
																																					.GetReadableFullName());
			}
		}

		/// <summary>
		///   Creates a default instance of the cache stats collector for the instance.
		/// </summary>
		/// <returns>an instance</returns>
		protected virtual ICacheStatsCollector CreateDefaultStatsCollector()
		{
			Contract.Ensures(Contract.Result<ICacheStatsCollector>() != null);
			return new BenignCacheStatsCollector(Name);
		}

		/// <summary>
		///   Invoked by the framework when eviction notifications arrive.
		/// </summary>
		/// <param name="reactor">the reactor</param>
		/// <param name="key">the key to be evicted</param>
		protected virtual void EvictionNotificationCallback(Reactor<TKey> reactor, TKey key)
		{
			if (!IsDisposed)
			{
				TValue removed;
				if (InternalTryRemove(key, out removed))
				{
					_stats.IncrementEvictions();
				}
			}
		}

		/// <summary>
		///   Provides subclasses the ability to resolve a group of missing items.
		///   Default behavior does nothing and returns false.
		/// </summary>
		/// <param name="keys">the item's keys</param>
		/// <param name="continuation">a continuation called upon completion</param>
		protected virtual void TryResolveManyMissingItems(IEnumerable<TKey> keys,
			Continuation<IEnumerable<Tuple<ResolutionKinds, TKey, TValue>>> continuation)
		{
			continuation(null, null);
		}

		/// <summary>
		///   Provides subclasses the ability to resolve missing items.
		///   Default behavior does nothing and returns false.
		/// </summary>
		/// <param name="key">the item's key</param>
		/// <param name="continuation">a continuation called upon completion</param>
		protected virtual void TryResolveMissingItem(TKey key, Continuation<Tuple<ResolutionKinds, TKey, TValue>> continuation)
		{
			continuation(null, null);
		}

		/// <summary>
		///   Updates statistics when items are resolved.
		/// </summary>
		/// <param name="stats">statistics collector</param>
		/// <param name="kind">kind of resolution</param>
		protected virtual void UpdateResolveCounters(ICacheStatsCollector stats, ResolutionKinds kind)
		{
			if (kind.HasFlag(ResolutionKinds.Resolved))
			{
				stats.IncrementHits(1);
			}
		}

		/// <summary>
		///   Performs dispose logic.
		/// </summary>
		/// <param name="disposing">whether called from Dispose method</param>
		/// <returns>true if disposed as a result of this call</returns>
		protected override bool PerformDispose(bool disposing)
		{
			Util.Dispose(ref _stats);
			if (_evictionMonitor != null)
			{
				_evictionManager.RemoveEvictionMonitor(_evictionMonitor.RegistrationKey);
				Util.Dispose(ref _evictionMonitor);
			}
			return true;
		}

		/// <summary>
		///   Removes the item without incrementing the remove counter.
		/// </summary>
		bool InternalTryRemove(TKey key, out TValue removed)
		{
			Item item;
			var result = _cache.TryRemove(key, out item);
			if (result)
			{
				removed = item.Remove(Interlocked.Increment(ref _revision));
				return true;
			}
			removed = default(TValue);
			return false;
		}

		void MostRecentlyUsed(Item item, bool reading)
		{
			var rev = reading ? Thread.VolatileRead(ref _revision) : Interlocked.Increment(ref _revision);
			item.UpdateRevision(rev);
			_lruQueue.Enqueue(new EvictionCandidate(item, rev));

			EvictionCandidate evictionCandidate;
			if (_lruQueue.Count > _cacheDepth && _lruQueue.TryDequeue(out evictionCandidate)
				&& evictionCandidate.RevisionBefore(rev - _cacheDepth))
			{
				_evictionQueue.Enqueue(new EvictionCandidate(evictionCandidate.Item, rev));
				if (_evictionQueue.Count > _evictionSynchronizedThreshold)
				{
					Interlocked.Increment(ref _evictorCount);
					ProcessEvictions(true);
				}
				else if (Interlocked.CompareExchange(ref _evictorCount, 1, 0) == 0)
				{
					_evictionManager.AddEvictionCallback(ProcessEvictions);
				}
			}
		}

		bool ProcessEvictions(bool isSynchronous)
		{
			int items = 0, evictions = 0;
			var reschedule = false;
			try
			{
				if (isSynchronous)
				{
					if (LogSink.ShouldTrace(this, TraceEventType.Verbose))
					{
						LogSink.OnTraceEvent(this, TraceEventType.Verbose,
																String.Concat(this.What, ".ProcessEvictions: synchronous worker started"));
					}
				}
				else
				{
					Interlocked.Increment(ref _activeEvictorCount);
				}
				EvictionCandidate candidate;
				while (_evictionQueue.TryDequeue(out candidate))
				{
					items += 1;
					var evict = candidate.ShouldEvict(_stats);
					if (evict.HasFlag(ShouldEvict.Yes))
					{
						TValue removed;
						if (InternalTryRemove(candidate.Item.Key, out removed))
						{
							_stats.IncrementEvictions();
							evictions++;
						}
					}
					if (isSynchronous && items > _evictionsPerSynchronizedWorker)
					{
						break;
					}
				}
			}
			finally
			{
				_stats.IncrementEvictions(evictions);
				if (LogSink.ShouldTrace(this, TraceEventType.Verbose))
				{
					LogSink.OnTraceEvent(this, TraceEventType.Verbose,
															String.Concat(this.What, "ProcessEvictions: worker done; processed ", items, " items and evicted ",
																						evictions));
				}
				int remaining = _evictionQueue.Count;
				if (isSynchronous)
				{
					Interlocked.Decrement(ref _evictorCount);
				}
				else
				{
					// Only reschedule when we are the last background thread observing
					// that there are remaining items.
					if (Interlocked.Decrement(ref _activeEvictorCount) == 0)
					{
						reschedule = remaining > 0;
						if (!reschedule)
						{
							Interlocked.Decrement(ref _evictorCount);
						}
					}
				}
			}

			return reschedule;
		}

		#region ICache<K,V> Members

		/// <summary>
		///   The cache's name. Must be unique within the process.
		/// </summary>
		public string Name { get; private set; }

		/// <summary>
		///   Asynchronously gets an item from the cache.
		/// </summary>
		/// <param name="key">the item's key</param>
		/// <param name="continuation">a continuation called upon completion</param>
		public void Get(TKey key, Continuation<Tuple<TKey, TValue>> continuation)
		{
			try
			{
				_stats.IncrementReadAttempts();
				Item item;
				if (_cache.TryGetValue(key, out item))
				{
					TValue existing = item.Value;
					if (!item.IsExpired(_stats) && existing != null)
					{
						_stats.IncrementHits(1);
						MostRecentlyUsed(item, true);
						continuation(null, Tuple.Create(key, existing));
						return;
					}
				}
			}
			catch (Exception e)
			{
				continuation(e, null);
				return;
			}

			Go.Parallel(() =>
			{
				try
				{
					// Second-chance... try to resolve...
					TryResolveMissingItem(key,
																(e, res) =>
																{
																	if (e != null)
																	{
																		continuation(e, null);
																	}
																	else
																	{
																		var result = default(Tuple<TKey, TValue>);
																		try
																		{
																			if (res != null)
																			{
																				UpdateResolveCounters(_stats, res.Item1);
																				if (res.Item1 == ResolutionKinds.Created)
																				{
																					// trust the newly resolved value...		
																					var item = new Item(key, res.Item3, Interlocked.Increment(ref _revision), _cacheItemLifespan);
																					_cache.AddOrUpdate(key, item, (k, r) => item);
																				}
																				if (res.Item1 != ResolutionKinds.None)
																				{
																					result = Tuple.Create(res.Item2, res.Item3);
																				}
																			}
																		}
																		catch (Exception ee)
																		{
																			continuation(ee, null);
																			return;
																		}
																		continuation(null, result);
																	}
																});
				}
				catch (Exception e)
				{
					continuation(e, null);
				}
			});
		}

		/// <summary>
		///   Asynchronously puts an item in the cache.
		/// </summary>
		/// <param name="key">the item's key</param>
		/// <param name="item">the item.</param>
		/// <param name="continuation">an optional continuation called upon completion</param>
		public void Put(TKey key, TValue item, Continuation<Tuple<TKey, TValue>> continuation)
		{
			try
			{
				var it = new Item(key, item, Interlocked.Increment(ref _revision), _cacheItemLifespan);
				_cache.AddOrUpdate(key, it, (k, v) => it);
				_stats.IncrementWrites();
				MostRecentlyUsed(it, false);
			}
			catch (Exception e)
			{
				if (continuation != null)
				{
					continuation(e, null);
				}
				return;
			}
			if (continuation != null)
			{
				continuation(null, Tuple.Create(key, item));
			}
		}

		/// <summary>
		///   Asynchronously removes an item from the cache.
		/// </summary>
		/// <param name="key">the item's key</param>
		/// <param name="continuation">an optional continuation called upon completion</param>
		public void Remove(TKey key, Continuation<Tuple<TKey, TValue>> continuation)
		{
			TValue res;
			try
			{
				if (InternalTryRemove(key, out res))
				{
					_stats.IncrementRemoves();
				}
			}
			catch (Exception e)
			{
				if (continuation != null)
				{
					continuation(e, null);
				}
				return;
			}
			if (continuation != null)
			{
				continuation(null, Tuple.Create(key, res));
			}
		}

		/// <summary>
		///   Asynchronously gets items from the cache.
		/// </summary>
		/// <param name="keys">keys of items to retrieve.</param>
		/// <param name="continuation">a continuation called upon completion</param>
		public void Get(IEnumerable<TKey> keys, Continuation<IEnumerable<Tuple<TKey, TValue>>> continuation)
		{
			Go.Parallel(() =>
			{
				var accum = new ConcurrentBag<Tuple<TKey, TValue>>();
				var missed = new List<TKey>();
				try
				{
					Parallel.ForEach(keys, k =>
					{
						Item item;
						_stats.IncrementReadAttempts();
						if (_cache.TryGetValue(k, out item))
						{
							TValue existing = item.Value;
							if (!item.IsExpired(_stats) && existing != null)
							{
								_stats.IncrementHits(1);
								MostRecentlyUsed(item, true);
								accum.Add(new Tuple<TKey, TValue>(k, existing));
							}
							else
							{
								missed.Add(k);
							}
						}
						else
						{
							missed.Add(k);
						}
					});
				}
				catch (Exception e)
				{
					continuation(e, null);
					return;
				}

				if (missed.Any())
				{
					// Second-chance... try to resolve...
					TryResolveManyMissingItems(missed, (e, res) =>
					{
						if (e != null)
						{
							continuation(e, null);
						}
						else
						{
							if (res != null)
							{
								try
								{
									foreach (var it in res)
									{
										UpdateResolveCounters(_stats, it.Item1);
										if (ResolutionKinds.Created == it.Item1)
										{
											// trust the newly resolved value...		
											var item = new Item(it.Item2, it.Item3, Interlocked.Increment(ref _revision), _cacheItemLifespan);
											_cache.AddOrUpdate(it.Item2, item, (k, r) => item);
											accum.Add(Tuple.Create(it.Item2, it.Item3));
										}
									}
								}
								catch (Exception ee)
								{
									continuation(ee, null);
									return;
								}
							}
							continuation(null, accum);
						}
					});
				}
				else
				{
					continuation(null, accum);
				}
			});
		}

		/// <summary>
		///   Asynchronously puts items in the cache.
		/// </summary>
		/// <param name="items">items to put into the cache.</param>
		/// <param name="continuation">a continuation called upon completion</param>
		public void Put(IEnumerable<Tuple<TKey, TValue>> items, Continuation<IEnumerable<Tuple<TKey, TValue>>> continuation)
		{
			Go.Parallel(() =>
			{
				var res = new ConcurrentBag<Tuple<TKey, TValue>>();
				try
				{
					Parallel.ForEach(items, it =>
					{
						var item = new Item(it.Item1, it.Item2, Interlocked.Increment(ref _revision), _cacheItemLifespan);
						_cache.AddOrUpdate(it.Item1, item, (k, v) => item);
						_stats.IncrementWrites();
						MostRecentlyUsed(item, false);
						res.Add(it);
					});
				}
				catch (Exception e)
				{
					if (continuation != null)
					{
						continuation(e, null);
					}
					return;
				}
				if (continuation != null)
				{
					continuation(null, res);
				}
			});
		}

		/// <summary>
		///   Asynchronously removes items from the cache.
		/// </summary>
		/// <param name="keys">keys of items to retrieve.</param>
		/// <param name="continuation">a continuation called upon completion</param>
		public void Remove(IEnumerable<TKey> keys, Continuation<IEnumerable<Tuple<TKey, TValue>>> continuation)
		{
			Go.Parallel(() =>
			{
				var res = new ConcurrentBag<Tuple<TKey, TValue>>();
				try
				{
					Parallel.ForEach(keys, k =>
					{
						TValue it;
						if (InternalTryRemove(k, out it))
						{
							res.Add(new Tuple<TKey, TValue>(k, it));
							_stats.IncrementRemoves();
						}
					});
				}
				catch (Exception e)
				{
					if (continuation != null)
					{
						continuation(e, null);
					}
					return;
				}
				if (continuation != null)
				{
					continuation(null, res);
				}
			});
		}

		/// <summary>
		///   Flushes all items from the cache.
		/// </summary>
		public void FlushAll()
		{
			_cache.Clear();
		}

		#endregion

		struct EvictionCandidate
		{
			readonly Item _item;
			readonly int _captureRevision;

			public EvictionCandidate(Item item, int revision)
			{
				Contract.Requires(item != null);
				_item = item;
				_captureRevision = revision;
			}

			internal Item Item { get { return _item; } }

			internal bool RevisionBefore(int revision)
			{
				return (_item.Revision < revision
					|| _item.Value == null);
			}

			internal ShouldEvict ShouldEvict(ICacheStatsCollector stats)
			{
				if (_item.Value == null)
				{
					return Cache.ShouldEvict.Cleared;
				}

				if (_item.Revision < _captureRevision)
				{
					return Cache.ShouldEvict.Revised;
				}

				if (_item.IsExpired(stats))
				{
					return Cache.ShouldEvict.Expired;
				}

				return Cache.ShouldEvict.No;
			}
		}

		class Item
		{
			int _expirationObserved;
			DateTime _expiry;
			readonly TKey _key;
			int _revision;
			TValue _value;

			public Item(TKey key, TValue value, int revision, TimeSpan lifespan)
			{
				this._key = key;
				this._value = value;
				this._revision = revision;
				this._expiry = DateTime.Now.Add(lifespan);
			}

			internal TKey Key { get { return _key; } }

			internal int Revision
			{
				get
				{
					Thread.MemoryBarrier();
					var value = _revision;
					Thread.MemoryBarrier();
					return value;
				}
			}

			internal TValue Value
			{
				get
				{
					Thread.MemoryBarrier();
					var value = _value;
					Thread.MemoryBarrier();
					return value;
				}
			}

			internal bool IsExpired(ICacheStatsCollector stats)
			{
				var revision = Thread.VolatileRead(ref _revision);
				var observed = Thread.VolatileRead(ref _expirationObserved);
				Thread.MemoryBarrier();
				var expire = _expiry;
				Thread.MemoryBarrier();
				var expired = DateTime.Now > expire;
				// test:
				// 1. the item is expired
				// 2. we are the first thread to observe it
				// 3. we are the first thread to mark it observed
				// 4. and no concurrent thread revised the item
				if (expired && observed == 0
					&& Interlocked.CompareExchange(ref _expirationObserved, 1, observed) == 0
					&& Thread.VolatileRead(ref _revision) == revision
					)
				{
					stats.IncrementExpired();
				}
				return expired;
			}

			internal TValue Remove(int revision)
			{
				Thread.MemoryBarrier();
				var existing = _value;
				Thread.MemoryBarrier();
				_value = default(TValue);
				_revision = revision;
				Thread.MemoryBarrier();
				return existing;
			}

			internal bool TryUpdate(TValue newValue, TValue expected, TimeSpan lifespan)
			{
				Thread.MemoryBarrier();
				var existing = _value;
				Thread.MemoryBarrier();

				var result = ReferenceEquals(existing, expected);
				if (result && Interlocked.CompareExchange(ref _value, newValue, existing) == existing)
				{
					var expire = DateTime.Now.Add(lifespan);
					Thread.MemoryBarrier();
					_expiry = expire;
					_expirationObserved = 0;
					Thread.MemoryBarrier();
					return true;
				}
				else
				{
					return false;
				}
			}

			internal void UpdateRevision(int revision)
			{
				Thread.VolatileWrite(ref _revision, revision);
			}
		}
	}
}